package com.dingtone.kafkaclient;

import java.util.Date;
import java.util.Properties;
import java.util.Random;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;

import com.dingtone.serialization.*;

import com.alibaba.fastjson.JSON;
import com.dingtone.Msg.Msg;

public class MsgProducer {

	public static void main(String[] args) {
		Random rnd = new Random();
		int events = 1000;
		//JsonSerializer<Msg> json = new JsonSerializer<Msg>();
		//JsonDeserializer jsonDeserializer = new JsonDeserializer();
		// 设置配置属性
		Properties props = new Properties();
		props.put("bootstrap.servers", "127.0.0.1:9092,127.0.0.1:9093,127.0.0.1:9094");
		//props.put("metadata.broker.list", "127.0.0.1:9092,127.0.0.1:9093,127.0.0.1:9094");
		props.put("serializer.class", "org.apache.kafka.common.serialization.StringSerializer");
		props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
		//props.put("value.serializer", json.getClass());
		//props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		//props.put("value.deserializer", jsonDeserializer.getClass());
		//props.put("group.id", "group-1");
		props.put("acks", "1"); // 可选配置，如果不配置，则使用默认的partitioner
		props.put("partitioner.class", "com.dingtone.kafkaclient.SimplePartitioner");

		// 创建producer
		Producer<String, Msg> producer = new KafkaProducer<String, Msg>(props,new StringSerializer(),new JsonSerializer<Msg>());
		// 产生并发送消息
		long start = System.currentTimeMillis();
		for (long i = 0; i < events; i++) {

			Msg msg = new Msg();
			msg.Time = new Date().getTime();
			msg.Ip = "192.168.2.110";
			msg.Content = "Hello World";
			msg.Ticket = i;
			String ip = "192.168.2." + i;
			ProducerRecord<String, Msg> data = new ProducerRecord<String, Msg>("page_visits0", ip,
					msg);
			producer.send(data, new Callback() {
				public void onCompletion(RecordMetadata metadata, Exception e) {
					if (e != null)
						e.printStackTrace();
					System.out.println("***[Producer]***send message: topic "+metadata.topic()+" partition "+metadata.partition() + metadata.offset() + "\n");
				}
			});
		}
		System.out.println("耗时:" + (System.currentTimeMillis() - start));
		// 关闭producer
		producer.close();
	}

//	public static String Serialize(Msg msg) {
//		return JSON.toJSONString(msg);
//	}
}
